package com.deng.rtc.timer;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import com.deng.rtc.client.LrtdcClient;
import com.deng.rtc.util.ConfigProperty;
import com.deng.rtc.util.Constants;
import com.deng.rtc.util.Murmurs;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.TimerTask;
import java.util.Map.Entry;

public class MqDoTimer extends TimerTask {
	// 实时接收MQ消息的本地队列
	protected List<ConcurrentLinkedQueue<String>> dataStreamList = new ArrayList<ConcurrentLinkedQueue<String>>();

	// AN 节点相关配置
	private String[] adminNodeHosts = Constants.adminNodeHosts.split(",");
	private int adminNodeNum;
	private ConcurrentHashMap<Integer, LrtdcClient> adminNodeClientMap;

	// google开源guava中的 murmurhash
	private Murmurs mmHash = new Murmurs();

	private long timerNum = 0;
	
	public MqDoTimer(){
		if(adminNodeHosts!=null){
			adminNodeNum = adminNodeHosts.length;
			adminNodeClientMap = new ConcurrentHashMap<Integer,LrtdcClient>();
			for(int i=0; i<adminNodeNum; i++){
				String[] tmpHostIp = adminNodeHosts[i].split(":");
				adminNodeClientMap.put(i, new LrtdcClient(tmpHostIp[0], Integer.parseInt(tmpHostIp[1])));
				dataStreamList.add(new ConcurrentLinkedQueue<String>());
			}
		}
	}
	// MQ所收集的每条数据通过murmurhash添加到相应AN节点ID所对应队列中
	public void parseMqText(String routKey, String mqText){
		int adminId = (int)(Murmurs.getGuavaMurmurs(routKey)%adminNodeNum);
		this.dataStreamList.get(adminId).add(mqText);
	}

	// 获取当前有效AN
	public List<Integer> getAbleAdminNodeIds(){
		List<Integer> ableJobNodeIds = new ArrayList<Integer>();
		for(Entry<Integer,LrtdcClient> item : adminNodeClientMap.entrySet()){
			if(item.getValue().getHealthStatus()==1){
				ableJobNodeIds.add(item.getKey());
			}
		}
		return ableJobNodeIds;
	}

	// CN资源调度器核心实现
	@Override
	public void run() {
		try{
			int tmpNum = -1;
			// 获取当前没个AN节点ID所对应本地队列的当前信息个数
			Map<Integer,Integer> dataQuNums = new HashMap<Integer,Integer>();
			for(int i=0; i<dataStreamList.size(); i++){
				tmpNum = dataStreamList.get(i).size();
				if(tmpNum>0){
					dataQuNums.put(i, tmpNum);
				}
			}
			// AN节点所对应的本地队列中只要有一个含有实时数据
			if(dataQuNums.size()>0){
				System.out.println(ConfigProperty.getCurDateTime()+" currently will be doing adminNode and logNum : "+dataQuNums);
				// 获取当前有效AN
				List<Integer> enableANIds = this.getAbleAdminNodeIds();
				if(enableANIds.size()>0){
					for(Entry<Integer,Integer> dqnItem : dataQuNums.entrySet()){
						// 启动新线程将当前AN节点ID所对应队列中数据提交给AN
						new Thread(new BatchDoLog(dqnItem.getKey(), dqnItem.getValue(), enableANIds)).start();
					}
				}else{
					System.out.println(ConfigProperty.getCurDateTime()+" currently there is no enable adminNode for receving stream logs");
				}
			}else{
				System.out.println(ConfigProperty.getCurDateTime()+"\t 当前时刻无消息 ");
			}
		} catch (Exception e) {
			System.err.println("用户行为日志收集错误 "+e.getMessage());
		}
		// 每提交1000批数据进行一次垃圾回收
		if(++timerNum%1000==0){
			//todo
			System.out.println("实时数据流按固定周期提交 " +timerNum+" 次");
			System.gc();
		}
	}

	// 内部类：完成Mini-Batch信息提交给AN
	protected class BatchDoLog implements Runnable{
		// AN节点ID
		private int adminId;
		// 当前日志队列中信息个数
		private int logNum;
		// 当前可用AN节点ID数组
		private List<Integer> enableAnIds;
		
		public BatchDoLog(int adminId, int logNum, List<Integer> enableAdminIds){
			this.adminId = adminId;
			this.logNum = logNum;
			this.enableAnIds = enableAdminIds;
		}

		private Random rand = new Random();
		
		@Override
		public void run() {
			List<String> tmpLogs = new LinkedList<String>();
			ConcurrentLinkedQueue<String> tmpQu = dataStreamList.get(adminId);
			for(int i=0;i<logNum;i++){
				tmpLogs.add(tmpQu.poll());
			}
			if(!enableAnIds.contains(adminId)){
				adminId = enableAnIds.get(rand.nextInt(enableAnIds.size()));
			}
			// CN提交当前批次实时数据给AN，即调用框架底层API实现
			adminNodeClientMap.get(adminId).addMqinfoToAdminQu(tmpLogs);
		}
	}

//	public static void main(String[] args){
//		MqDoTimer mdt = new MqDoTimer();
//		mdt.test();
//	}
	
}
